use crate::block::{Block, BlockIter};
use crate::blockhandle::BlockHandle;
use crate::cache::{self, Cache};
use crate::cmp::InternalKeyCmp;
use crate::env::RandomAccess;
use crate::error::{self, err, Result};
use crate::filter;
use crate::filter_block::FilterBlockReader;
use crate::key_types::InternalKey;
use crate::options::Options;
use crate::table_block;
use crate::table_builder::{self, Footer};
use crate::types::{current_key_val, LdbIterator, Shared};

use std::cmp::Ordering;
use std::rc::Rc;

use bytes::Bytes;
use integer_encoding::FixedIntWriter;

/// Reads the table footer.
fn read_footer(f: &dyn RandomAccess, size: usize) -> Result<Footer> {
    let mut buf = vec![0; table_builder::FULL_FOOTER_LENGTH];
    f.read_at(size - table_builder::FULL_FOOTER_LENGTH, &mut buf)?;
    match Footer::decode(&buf) {
        Some(ok) => Ok(ok),
        None => err(
            error::StatusCode::Corruption,
            &format!("Couldn't decode damaged footer {:?}", &buf),
        ),
    }
}

/// 这个 Table 结构体是 LSM 树存储引擎中SSTable 文件的内存映射与访问接口，封装了对磁盘上 SSTable 文件的读取、解析和缓存逻辑，是查询时访问 SSTable 数据的核心工具。
#[derive(Clone)]
pub struct Table {
    /// 随机访问文件接口，用于读取 SSTable 文件的任意位置（支持磁盘文件或缓存）。
    file: Rc<Box<dyn RandomAccess>>,
    /// SSTable 文件的总大小（字节），用于边界检查和定位页脚等固定位置结构。
    file_size: usize,
    /// 块缓存的唯一标识，用于在全局块缓存中定位当前 SSTable 的缓存数据。
    cache_id: cache::CacheID,

    /// 数据库配置，包含块大小、比较器、压缩算法等影响 SSTable 解析的参数。
    opt: Options,
    /// 块缓存引用，用于缓存解析后的 data block（数据块），减少重复磁盘 I/O。
    block_cache: Shared<Cache<Block>>,

    /// SSTable 页脚，存储索引块、过滤器块的位置和大小等元数据（文件末尾固定结构）。
    footer: Footer,
    /// 索引块，存储数据块的边界键和偏移量，用于快速定位目标键所在的数据块。
    indexblock: Block,
    /// 过滤器块读取器（如布隆过滤器），用于快速判断键是否可能存在于 SSTable 中，减少无效数据块查询。
    filters: Option<FilterBlockReader>,
}

impl Table {
    /// Creates a new table reader operating on unformatted keys (i.e., UserKey).
    fn new_raw(
        opt: Options,
        block_cache: Shared<Cache<Block>>,
        file: Rc<Box<dyn RandomAccess>>,
        size: usize,
    ) -> Result<Table> {
        let footer = read_footer(file.as_ref().as_ref(), size)?;
        let indexblock =
            table_block::read_table_block(opt.clone(), file.as_ref().as_ref(), &footer.index)?;
        let metaindexblock =
            table_block::read_table_block(opt.clone(), file.as_ref().as_ref(), &footer.meta_index)?;

        let filter_block_reader =
            Table::read_filter_block(&metaindexblock, file.as_ref().as_ref(), &opt)?;
        let cache_id = block_cache.borrow_mut().new_cache_id();

        Ok(Table {
            file,
            file_size: size,
            cache_id,
            opt,
            block_cache,
            footer,
            filters: filter_block_reader,
            indexblock,
        })
    }

    fn read_filter_block(
        metaix: &Block,
        file: &dyn RandomAccess,
        options: &Options,
    ) -> Result<Option<FilterBlockReader>> {
        // Open filter block for reading
        let filter_name = format!("filter.{}", options.filter_policy.name())
            .as_bytes()
            .to_vec();

        let mut metaindexiter = metaix.iter();
        metaindexiter.seek(&filter_name);

        if let Some((_key, val)) = current_key_val(&metaindexiter) {
            let fbl = BlockHandle::decode(&val);
            let filter_block_location = match fbl {
                None => {
                    return err(
                        error::StatusCode::Corruption,
                        &format!("Couldn't decode corrupt blockhandle {:?}", &val),
                    )
                }
                Some(ok) => ok.0,
            };
            if filter_block_location.size() > 0 {
                return Ok(Some(table_block::read_filter_block(
                    file,
                    &filter_block_location,
                    options.filter_policy.clone(),
                )?));
            }
        }
        Ok(None)
    }

    /// Creates a new table reader operating on internal keys (i.e., InternalKey). This means that
    /// a different comparator (internal_key_cmp) and a different filter policy
    /// (InternalFilterPolicy) are used.
    pub fn new(
        mut opt: Options,
        block_cache: Shared<Cache<Block>>,
        file: Rc<Box<dyn RandomAccess>>,
        size: usize,
    ) -> Result<Table> {
        opt.cmp = Rc::new(Box::new(InternalKeyCmp(opt.cmp.clone())));
        opt.filter_policy = Rc::new(Box::new(filter::InternalFilterPolicy::new(
            opt.filter_policy,
        )));
        Table::new_raw(opt, block_cache, file, size)
    }

    /// block_cache_handle creates a CacheKey for a block with a given offset to be used in the block cache.
    ///
    /// 这个 block_cache_handle 函数用于生成数据块在全局缓存中的唯一标识（CacheKey），确保不同 SSTable 中的数据块即使偏移量相同，也能在缓存中被正确区分。
    /// 这是实现跨多个 SSTable 共享块缓存的关键。
    ///
    /// block_cache_handle 的核心价值在于生成全局唯一的缓存键，解决了两个关键问题：
    ///
    /// 1.避免不同 SSTable 的块冲突
    ///
    /// 不同 SSTable 文件中可能存在偏移量相同的数据块（例如两个文件的 1024 字节位置都有一个数据块），通过加入 cache_id（SSTable 唯一标识），确保它们在缓存中被视为不同的键，不会相互覆盖。
    ///
    /// 2.精确定位缓存中的块
    ///
    /// 缓存键同时包含 SSTable 标识和块偏移量，使得后续读取同一数据块时，能准确匹配到缓存中的对应条目，保证缓存机制的正确性。
    fn block_cache_handle(&self, block_off: usize) -> cache::CacheKey {
        let mut dst = [0; 2 * 8]; // 共 16 字节（2 个 64 位整数）
        // 将当前 SSTable 的 cache_id（每个 SSTable 实例的唯一标识）写入数组的前 8 字节；
        // write_fixedint 用于将整数以固定字节长度（这里是 8 字节）写入字节切片。
        (&mut dst[..8])
            .write_fixedint(self.cache_id)
            .expect("error writing to vec");
        // 将数据块在 SSTable 文件中的偏移量（block_off）转换为 64 位整数后，写入数组的后 8 字节。
        (&mut dst[8..])
            .write_fixedint(block_off as u64)
            .expect("error writing to vec");
        dst
    }

    /// Read a block from the current table at `location`, and cache it in the options' block cache.
    ///
    /// 这个 read_block 方法是 SSTable 读取数据块的核心工具函数，负责从缓存或磁盘中获取指定位置的数据块（Block），并将读取到的块存入缓存以加速后续访问。
    /// 它是平衡磁盘 I/O 和内存缓存的关键实现。
    fn read_block(&self, location: &BlockHandle) -> Result<Block> {
        // 生成缓存键：block_cache_handle(location.offset()) 基于数据块的偏移量生成唯一的缓存键（cachekey），用于标识该数据块；
        let cachekey = self.block_cache_handle(location.offset());
        // 检查缓存：通过 block_cache.borrow_mut().get(&cachekey) 查看块缓存中是否已存在该数据块：
        // 若存在（缓存命中），直接返回缓存中的块副本（block.clone()），避免磁盘 I/O；
        // 若不存在（缓存未命中），继续执行磁盘读取流程。
        if let Some(block) = self.block_cache.borrow_mut().get(&cachekey) {
            return Ok(block.clone());
        }

        // Two times as_ref(): First time to get a ref from Rc<>, then one from Box<>.
        // 调用 read_table_block 函数，根据 location 中的偏移量和大小，从 SSTable 文件（self.file）中读取原始数据，并解析为 Block 结构体（包含数据块的键值对和索引信息）；
        // 传入 self.opt（数据库配置）用于解析时的参数（如块大小、压缩算法等）。
        let b = table_block::read_table_block(self.opt.clone(), self.file.as_ref().as_ref(), location)?;

        // insert a cheap copy (Rc).
        // 读取成功后，将数据块克隆一份存入块缓存（block_cache），键为之前生成的 cachekey；
        // 后续再访问同一数据块时，可直接从缓存获取，提升性能。
        self.block_cache.borrow_mut().insert(&cachekey, b.clone());

        Ok(b)
    }

    /// Returns the offset of the block that contains `key`.
    pub fn approx_offset_of(&self, key: &[u8]) -> usize {
        let mut iter = self.indexblock.iter();

        iter.seek(key);

        if let Some((_, val)) = current_key_val(&iter) {
            let location = BlockHandle::decode(&val).unwrap().0;
            return location.offset();
        }

        self.footer.meta_index.offset()
    }

    /// Iterators read from the file; thus only one iterator can be borrowed (mutably) per scope
    pub fn iter(&self) -> TableIterator {
        TableIterator {
            current_block: None,
            current_block_off: 0,
            index_block: self.indexblock.iter(),
            table: self.clone(),
        }
    }

    /// Retrieve next-biggest entry for key from table. This function uses the attached filters, so
    /// is better suited if you frequently look for non-existing values (as it will detect the
    /// non-existence of an entry in a block without having to load the block).
    ///
    /// The caller must check if the returned key, which is the raw key (not e.g. the user portion
    /// of an InternalKey) is acceptable (e.g. correct value type or sequence number), as it may
    /// not be an exact match for key.
    ///
    /// This is done this way because some key types, like internal keys, will not result in an
    /// exact match; it depends on other comparators than the one that the table reader knows
    /// whether a match is acceptable.
    ///
    /// 这个 get 方法是 SSTable（有序字符串表）的核心查询接口，用于根据内部键（InternalKey）从 SSTable 中查找对应的键值对。
    /// 它通过多层索引和缓存机制，高效定位目标数据，是 LSM 树存储引擎读取数据的关键环节。
    ///
    /// 这个 get 方法是 SSTable 的 **“数据查询引擎”**，其核心价值在于：
    ///
    /// 多层索引加速：通过索引块快速定位数据块，避免扫描整个 SSTable；
    ///
    /// 过滤器优化：利用布隆过滤器等提前排除不存在的键，减少磁盘访问；
    ///
    /// 缓存利用：通过 read_block 优先从缓存获取数据块，提升查询性能；
    ///
    /// 高效匹配：通过迭代器的 seek 方法（通常基于二分查找）在数据块内快速定位目标键。
    pub fn get(&self, key: InternalKey<'_>) -> Result<Option<(Bytes, Bytes)>> {
        // 定位目标数据块（通过索引块） BlockIter
        // 为索引块创建迭代器 index_iter = self.indexblock.iter()，其中 Block::iter() 返回 BlockIter 实例（代码片段见 leveldb-rs/src/block.rs 中 Block::iter 方法）。
        // 这里 BlockIter 进
        let mut index_iter = self.indexblock.iter();
        // 通过 index_iter.seek(key) 在索引块中定位到第一个边界键大于等于目标键的条目；
        // 若目标键小于该边界键，说明目标键可能在对应的 data block 中，解析出数据块的位置（handle）；
        // 否则，目标键不在当前 SSTable 中，返回 None。
        // 调用 index_iter.seek(key) 定位到目标数据块的 BlockHandle（偏移量和大小），这里的 seek 是 BlockIter 的方法，会打印 ----------block seek---------- 等日志。
        index_iter.seek(key);

        let handle;
        // 索引块（indexblock）：SSTable 的索引块存储了每个数据块的边界键（last_in_block）和数据块的位置信息（BlockHandle，包含偏移量和大小）。 BlockIter
        // 这里 BlockIter 进
        if let Some((last_in_block, h)) = current_key_val(&index_iter) {
            if self.opt.cmp.cmp(key, &last_in_block) == Ordering::Less {
                handle = BlockHandle::decode(&h).unwrap().0;
            } else {
                return Ok(None);
            }
        } else {
            return Ok(None);
        }

        // found correct block.

        // Check bloom (or whatever) filter
        // 过滤不存在的键（通过过滤器）
        // 过滤器（如布隆过滤器）：用于快速判断键是否 “可能” 存在于数据块中，减少无效的磁盘 I/O。
        // 作用：如果过滤器判断键一定不存在，则直接返回 None，无需读取数据块；只有 “可能存在” 时才继续查询
        if let Some(ref filters) = self.filters {
            if !filters.key_may_match(handle.offset(), key) {
                return Ok(None);
            }
        }

        // Read block (potentially from cache)
        // 读取数据块：根据 handle 中的偏移量和大小，从磁盘或块缓存（block_cache）中读取目标数据块（tb）；
        let tb = self.read_block(&handle)?;
        // 创建 数据块迭代器：通过数据块的迭代器遍历其中的键值对。
        let mut iter = tb.iter();

        // Go to entry and check if it's the wanted entry.
        // 在数据块中定位：通过 iter.seek(key) 找到数据块中第一个大于等于目标键的条目；
        // 匹配判断：若找到的键与目标键相等（或根据比较器判断匹配），返回键值对；否则返回 None。
        // 调用 data_iter.seek(key) 在数据块中精确查找目标键，过程中会调用 BlockIter 的 seek、advance、valid 等方法，对应日志中的 ----------block advance----------、----------block valid---------- 等输出。
        iter.seek(key);
        if let Some((k, v)) = current_key_val(&iter) {
            if self.opt.cmp.cmp(&k, key) >= Ordering::Equal {
                return Ok(Some((k.into(), v.into())));
            }
        }
        Ok(None)
    }
}

/// This iterator is a "TwoLevelIterator"; it uses an index block in order to get an offset hint
/// into the data blocks.
pub struct TableIterator {
    // A TableIterator is independent of its table (on the syntax level -- it doesn't know its
    // Table's lifetime). This is mainly required by the dynamic iterators used everywhere, where a
    // lifetime makes things like returning an iterator from a function neigh-impossible.
    //
    // Instead, reference-counted pointers and locks inside the Table ensure that all
    // TableIterators still share a table.
    table: Table,
    current_block: Option<BlockIter>,
    current_block_off: usize,
    index_block: BlockIter,
}

impl TableIterator {
    // Skips to the entry referenced by the next entry in the index block.
    // This is called once a block has run out of entries.
    // Err means corruption or I/O error; Ok(true) means a new block was loaded; Ok(false) means
    // tht there's no more entries.
    fn skip_to_next_entry(&mut self) -> Result<bool> {
        if let Some((_key, val)) = self.index_block.next() {
            self.load_block(&val).map(|_| true)
        } else {
            Ok(false)
        }
    }

    // Load the block at `handle` into `self.current_block`
    fn load_block(&mut self, handle: &[u8]) -> Result<()> {
        let (new_block_handle, _) = match BlockHandle::decode(handle) {
            None => {
                return err(
                    error::StatusCode::Corruption,
                    "Couldn't decode corrupt block handle",
                )
            }
            Some(ok) => ok,
        };
        let block = self.table.read_block(&new_block_handle)?;

        self.current_block = Some(block.iter());
        self.current_block_off = new_block_handle.offset();

        Ok(())
    }
}

impl LdbIterator for TableIterator {
    fn advance(&mut self) -> bool {
        println!("----------table advance----------");
        // Uninitialized case.
        if self.current_block.is_none() {
            match self.skip_to_next_entry() {
                Ok(true) => return self.advance(),
                Ok(false) => {
                    self.reset();
                    return false;
                }
                // try next block from index, this might be corruption
                Err(_) => return self.advance(),
            }
        }

        // Initialized case -- does the current block have more entries?
        if let Some(ref mut cb) = self.current_block {
            if cb.advance() {
                return true;
            }
        }

        // If the current block is exhausted, try loading the next block.
        self.current_block = None;
        match self.skip_to_next_entry() {
            Ok(true) => self.advance(),
            Ok(false) => {
                self.reset();
                false
            }
            // try next block, this might be corruption
            Err(_) => self.advance(),
        }
    }

    // A call to valid() after seeking is necessary to ensure that the seek worked (e.g., no error
    // while reading from disk)
    fn seek(&mut self, to: &[u8]) {
        println!("----------table seek----------");
        // first seek in index block, rewind by one entry (so we get the next smaller index entry),
        // then set current_block and seek there
        self.index_block.seek(to);

        // It's possible that this is a seek past-last; reset in that case.
        if let Some((past_block, handle)) = current_key_val(&self.index_block) {
            if self.table.opt.cmp.cmp(to, &past_block) <= Ordering::Equal {
                // ok, found right block: continue
                if let Ok(()) = self.load_block(&handle) {
                    // current_block is always set if load_block() returned Ok.
                    self.current_block.as_mut().unwrap().seek(to);
                    return;
                }
            }
        }
        // Reached in case of failure.
        self.reset();
    }

    fn prev(&mut self) -> bool {
        println!("----------table prev----------");
        // happy path: current block contains previous entry
        if let Some(ref mut cb) = self.current_block {
            if cb.prev() {
                return true;
            }
        }

        // Go back one block and look for the last entry in the previous block
        if self.index_block.prev() {
            if let Some((_, handle)) = current_key_val(&self.index_block) {
                if self.load_block(&handle).is_ok() {
                    self.current_block.as_mut().unwrap().seek_to_last();
                    self.current_block.as_ref().unwrap().valid()
                } else {
                    self.reset();
                    false
                }
            } else {
                false
            }
        } else {
            false
        }
    }

    fn reset(&mut self) {
        println!("----------table reset----------");
        self.index_block.reset();
        self.current_block = None;
    }

    // This iterator is special in that it's valid even before the first call to advance(). It
    // behaves correctly, though.
    fn valid(&self) -> bool {
        println!("----------table valid----------");
        self.current_block.is_some() && (self.current_block.as_ref().unwrap().valid())
    }

    fn current(&self) -> Option<(Bytes, Bytes)> {
        println!("----------table current----------");
        if let Some(ref cb) = self.current_block {
            cb.current()
        } else {
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use std::cell::RefCell;

    use crate::compressor::CompressorId;
    use crate::filter::BloomPolicy;
    use crate::key_types::LookupKey;
    use crate::table_builder::TableBuilder;
    use crate::test_util::{test_iterator_properties, LdbIteratorIter};
    use crate::types::{current_key_val, share, LdbIterator};
    use crate::{block, compressor, options};

    use super::*;

    fn build_data() -> Vec<(&'static str, &'static str)> {
        vec![
            // block 1
            ("abc", "def"),
            ("abd", "dee"),
            ("bcd", "asa"),
            // block 2
            ("bsr", "a00"),
            ("xyz", "xxx"),
            ("xzz", "yyy"),
            // block 3
            ("zzz", "111"),
        ]
    }

    // Build a table containing raw keys (no format). It returns (vector, length) for convenience
    // reason, a call f(v, v.len()) doesn't work for borrowing reasons.
    fn build_table(data: Vec<(&'static str, &'static str)>) -> (Vec<u8>, usize) {
        let mut d = Vec::with_capacity(512);
        let mut opt = options::for_test();
        opt.block_restart_interval = 2;
        opt.block_size = 32;
        opt.compressor = compressor::SnappyCompressor::ID;

        {
            // Uses the standard comparator in opt.
            let mut b = TableBuilder::new_raw(opt, &mut d);

            for &(k, v) in data.iter() {
                b.add(k.as_bytes(), v.as_bytes()).unwrap();
            }

            b.finish().unwrap();
        }

        let size = d.len();
        (d, size)
    }

    // Build a table containing keys in InternalKey format.
    fn build_internal_table() -> (Vec<u8>, usize) {
        let mut d = Vec::with_capacity(512);
        let mut opt = options::for_test();
        opt.block_restart_interval = 1;
        opt.block_size = 32;
        opt.filter_policy = Rc::new(Box::new(BloomPolicy::new(4)));

        let mut i = 1_u64;
        let data: Vec<(Vec<u8>, &'static str)> = build_data()
            .into_iter()
            .map(|(k, v)| {
                i += 1;
                (LookupKey::new(k.as_bytes(), i).internal_key().to_vec(), v)
            })
            .collect();

        {
            // Uses InternalKeyCmp
            let mut b = TableBuilder::new(opt, &mut d);

            for (k, v) in data.iter() {
                b.add(k.as_slice(), v.as_bytes()).unwrap();
            }

            b.finish().unwrap();
        }

        let size = d.len();

        (d, size)
    }

    fn wrap_buffer(src: Vec<u8>) -> Rc<Box<dyn RandomAccess>> {
        Rc::new(Box::new(src))
    }

    #[test]
    fn test_table_approximate_offset() {
        let (src, size) = build_table(build_data());
        let mut opt = options::for_test();
        let bc = share(Cache::new(128));
        opt.block_size = 32;
        let table = Table::new_raw(opt, bc, wrap_buffer(src), size).unwrap();
        let mut iter = table.iter();

        let expected_offsets = [0, 0, 0, 44, 44, 44, 89];
        for (i, (k, _)) in LdbIteratorIter::wrap(&mut iter).enumerate() {
            assert_eq!(expected_offsets[i], table.approx_offset_of(&k));
        }

        // Key-past-last returns offset of metaindex block.
        assert_eq!(137, table.approx_offset_of("{aa".as_bytes()));
    }

    #[test]
    fn test_table_block_cache_use() {
        let (src, size) = build_table(build_data());
        let bc = share(Cache::new(128));
        let mut opt = options::for_test();
        opt.block_size = 32;

        {
            let table = Table::new_raw(opt.clone(), bc.clone(), wrap_buffer(src), size).unwrap();
            let mut iter = table.iter();

            // index/metaindex blocks are not cached. That'd be a waste of memory.
            assert_eq!(bc.borrow().count(), 0);
            iter.next();
            assert_eq!(bc.borrow().count(), 1);
            // This may fail if block parameters or data change. In that case, adapt it.
            iter.next();
            iter.next();
            iter.next();
            iter.next();
            assert_eq!(bc.borrow().count(), 2);
        }

        println!(
            "weak = {}, strong = {}",
            std::rc::Rc::<RefCell<cache::Cache<block::Block>>>::weak_count(&bc),
            std::rc::Rc::<RefCell<cache::Cache<block::Block>>>::strong_count(&bc)
        );
    }

    #[test]
    fn test_table_iterator_fwd_bwd() {
        let (src, size) = build_table(build_data());
        let data = build_data();
        let bc = share(Cache::new(128));

        let table = Table::new_raw(options::for_test(), bc, wrap_buffer(src), size).unwrap();
        let mut iter = table.iter();
        let mut i = 0;

        while let Some((k, v)) = iter.next() {
            assert_eq!(
                (data[i].0.as_bytes(), data[i].1.as_bytes()),
                (k.as_ref(), v.as_ref())
            );
            i += 1;
        }

        assert_eq!(i, data.len());
        assert!(!iter.valid());

        // Go forward again, to last entry.
        while let Some((key, _)) = iter.next() {
            if key.as_slice() == b"zzz" {
                break;
            }
        }

        assert!(iter.valid());
        // backwards count
        let mut j = 0;

        while iter.prev() {
            if let Some((k, v)) = current_key_val(&iter) {
                j += 1;
                assert_eq!(
                    (
                        data[data.len() - 1 - j].0.as_bytes(),
                        data[data.len() - 1 - j].1.as_bytes()
                    ),
                    (k.as_ref(), v.as_ref())
                );
            } else {
                break;
            }
        }

        // expecting 7 - 1, because the last entry that the iterator stopped on is the last entry
        // in the table; that is, it needs to go back over 6 entries.
        assert_eq!(j, 6);
    }

    #[test]
    fn test_table_iterator_filter() {
        let (src, size) = build_table(build_data());
        let bc = share(Cache::new(128));

        let table = Table::new_raw(options::for_test(), bc, wrap_buffer(src), size).unwrap();
        assert!(table.filters.is_some());
        let filter_reader = table.filters.clone().unwrap();
        let mut iter = table.iter();
        while let Some((k, _)) = iter.next() {
            assert!(filter_reader.key_may_match(iter.current_block_off, &k));
            assert!(!filter_reader.key_may_match(iter.current_block_off, b"somerandomkey"));
        }
    }

    #[test]
    fn test_table_iterator_state_behavior() {
        let (src, size) = build_table(build_data());
        let bc = share(Cache::new(128));

        let table = Table::new_raw(options::for_test(), bc, wrap_buffer(src), size).unwrap();
        let mut iter = table.iter();

        // behavior test

        // See comment on valid()
        assert!(!iter.valid());
        assert!(current_key_val(&iter).is_none());
        assert!(!iter.prev());

        assert!(iter.advance());
        let first = current_key_val(&iter);
        assert!(iter.valid());
        assert!(current_key_val(&iter).is_some());

        assert!(iter.advance());
        assert!(iter.prev());
        assert!(iter.valid());

        iter.reset();
        assert!(!iter.valid());
        assert!(current_key_val(&iter).is_none());
        assert_eq!(first, iter.next());
    }

    #[test]
    fn test_table_iterator_behavior_standard() {
        let mut data = build_data();
        data.truncate(4);
        let (src, size) = build_table(data);
        let bc = share(Cache::new(128));
        let table = Table::new_raw(options::for_test(), bc, wrap_buffer(src), size).unwrap();
        test_iterator_properties(table.iter());
    }

    #[test]
    fn test_table_iterator_values() {
        let (src, size) = build_table(build_data());
        let data = build_data();
        let bc = share(Cache::new(128));

        let table = Table::new_raw(options::for_test(), bc, wrap_buffer(src), size).unwrap();
        let mut iter = table.iter();
        let mut i = 0;

        iter.next();
        iter.next();

        // Go back to previous entry, check, go forward two entries, repeat
        // Verifies that prev/next works well.
        loop {
            iter.prev();

            if let Some((k, v)) = current_key_val(&iter) {
                assert_eq!(
                    (data[i].0.as_bytes(), data[i].1.as_bytes()),
                    (k.as_ref(), v.as_ref())
                );
            } else {
                break;
            }

            i += 1;
            if iter.next().is_none() || iter.next().is_none() {
                break;
            }
        }

        // Skipping the last value because the second next() above will break the loop
        assert_eq!(i, 6);
    }

    #[test]
    fn test_table_iterator_seek() {
        let (src, size) = build_table(build_data());
        let bc = share(Cache::new(128));

        let table = Table::new_raw(options::for_test(), bc, wrap_buffer(src), size).unwrap();
        let mut iter = table.iter();

        iter.seek(b"bcd");
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter),
            Some((b"bcd".to_vec(), b"asa".to_vec()))
        );
        iter.seek(b"abc");
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter),
            Some((b"abc".to_vec(), b"def".to_vec()))
        );

        // Seek-past-last invalidates.
        iter.seek("{{{".as_bytes());
        assert!(!iter.valid());
        iter.seek(b"bbb");
        assert!(iter.valid());
    }

    #[test]
    fn test_table_get() {
        let (src, size) = build_table(build_data());
        let bc = share(Cache::new(128));

        let table =
            Table::new_raw(options::for_test(), bc.clone(), wrap_buffer(src), size).unwrap();
        let table2 = table.clone();

        let mut _iter = table.iter();
        // Test that all of the table's entries are reachable via get()
        for (k, v) in LdbIteratorIter::wrap(&mut _iter) {
            let r = table2.get(&k);
            assert_eq!(Ok(Some((k.into(), v.into()))), r);
        }

        assert_eq!(bc.borrow().count(), 3);

        // test that filters work and don't return anything at all.
        assert!(table.get(b"aaa").unwrap().is_none());
        assert!(table.get(b"aaaa").unwrap().is_none());
        assert!(table.get(b"aa").unwrap().is_none());
        assert!(table.get(b"abcd").unwrap().is_none());
        assert!(table.get(b"abb").unwrap().is_none());
        assert!(table.get(b"zzy").unwrap().is_none());
        assert!(table.get(b"zz1").unwrap().is_none());
        assert!(table.get("zz{".as_bytes()).unwrap().is_none());
    }

    // This test verifies that the table and filters work with internal keys. This means:
    // The table contains keys in InternalKey format and it uses a filter wrapped by
    // InternalFilterPolicy.
    // All the other tests use raw keys that don't have any internal structure; this is fine in
    // general, but here we want to see that the other infrastructure works too.
    #[test]
    fn test_table_internal_keys() {
        use crate::key_types::LookupKey;

        let (src, size) = build_internal_table();

        let bc = share(Cache::new(128));
        let table = Table::new(options::for_test(), bc, wrap_buffer(src), size).unwrap();
        let filter_reader = table.filters.clone().unwrap();

        // Check that we're actually using internal keys
        let mut _iter = table.iter();
        for (ref k, ref v) in LdbIteratorIter::wrap(&mut _iter) {
            assert_eq!(k.len(), 3 + 8);
            let (result_k, result_v) = table.get(k).unwrap().unwrap();
            assert_eq!(
                (k.to_vec(), v.to_vec()),
                (result_k.to_vec(), result_v.to_vec())
            );
        }

        assert!(table
            .get(LookupKey::new(b"abc", 1000).internal_key())
            .unwrap()
            .is_some());

        let mut iter = table.iter();

        while let Some((k, _)) = iter.next() {
            let lk = LookupKey::new(&k, 123);
            let userkey = lk.user_key();

            assert!(filter_reader.key_may_match(iter.current_block_off, userkey));
            assert!(!filter_reader.key_may_match(iter.current_block_off, b"somerandomkey"));
        }
    }

    #[test]
    fn test_table_reader_checksum() {
        let bc = share(Cache::new(128));
        let (mut src, size) = build_table(build_data());

        src[10] += 1;

        let table = Table::new_raw(options::for_test(), bc, wrap_buffer(src), size).unwrap();

        assert!(table.filters.is_some());
        assert_eq!(table.filters.as_ref().unwrap().num(), 1);

        {
            let mut _iter = table.iter();
            let iter = LdbIteratorIter::wrap(&mut _iter);
            // first block is skipped
            assert_eq!(iter.count(), 4);
        }

        {
            let mut _iter = table.iter();
            let iter = LdbIteratorIter::wrap(&mut _iter);

            for (k, _) in iter {
                if k == build_data()[5].0.as_bytes() {
                    return;
                }
            }

            panic!("Should have hit 5th record in table!");
        }
    }
}
